package com.pingan.haofang.searchcloud.common.redis;

import com.pingan.haofang.searchcloud.common.utils.RedisUtils;
import com.pingan.haofang.searchcloud.index.constants.IndexBuildConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.List;

/**
 * TODO: 请添加描述
 *
 * @author liujunjun380
 * @date 2018/11/19
 * @since 1.0.0
 */
public class IncRedisQueue<T> extends Thread {

    static Logger LOG = LoggerFactory.getLogger(IncRedisQueue.class);

    private RedisTemplate redisTemplate;
    private String key;
    private BoundListOperations<String, T> listOperations;
    private RedisQueueListener<T> listener;
    /**
     * 每次消息处理个数
     */
    private int batchSize = 10;

    //锁过期时间ms
    private int expireTime = 1000;


    public IncRedisQueue(String key, RedisTemplate redisTemplate, RedisQueueListener<T> listener) {
        super(key);
        this.listener = listener;
        this.key = key;
        this.redisTemplate = redisTemplate;
        listOperations = redisTemplate.boundListOps(key);
    }

    @Override
    public void run() {
        //检查全量锁
        Boolean lock = redisTemplate.opsForHash().hasKey(IndexBuildConstants.LOCK, key.split(":")[1]);
        if (lock) {
            LOG.info("index build lock exists, key:{}, waiting for release", key);
            return;
        }

        //消费队列消息
        long size = listOperations.size() >= batchSize ? batchSize : listOperations.size();
        if (size > 0) {
            String distributeValue = String.valueOf(System.currentTimeMillis() + expireTime);
            String distributeKey = key + ":distribute";
            //加锁
            LOG.info("tryGetDistributedLock,key:{},value:{}", distributeKey, distributeValue);
            if (!RedisUtils.tryGetDistributedLock(redisTemplate, distributeKey, distributeValue)) {
                return;
            }
            List<T> values = listOperations.range(0, size - 1);
            listOperations.trim(size, -1);
            //释放锁
            RedisUtils.releaseDistributedLock(redisTemplate, distributeKey, distributeValue);
            LOG.info("releaseDistributedLock,key:{},value:{}", distributeKey, distributeValue);
            for (T value : values) {
                //逐个执行
                if (value != null) {
                    try {
                        listener.onMessage(value);
                    } catch (Exception e) {
                        LOG.error(e.getMessage(), e);
                    }
                }
            }
        }
    }

    public RedisTemplate getRedisTemplate() {
        return redisTemplate;
    }

    public IncRedisQueue setRedisTemplate(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        return this;
    }

    public String getKey() {
        return key;
    }

    public IncRedisQueue setKey(String key) {
        this.key = key;
        return this;
    }

    public RedisQueueListener<T> getListener() {
        return listener;
    }

    public IncRedisQueue setListener(RedisQueueListener<T> listener) {
        this.listener = listener;
        return this;
    }

    public int getBatchSize() {
        return batchSize;
    }

    public IncRedisQueue setBatchSize(int batchSize) {
        this.batchSize = batchSize;
        return this;
    }

    public int getExpireTime() {
        return expireTime;
    }

    public IncRedisQueue setExpireTime(int expireTime) {
        this.expireTime = expireTime;
        return this;
    }
}
